//package njuflink;
//
//import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
//
//import java.sql.PreparedStatement;
//import java.sql.ResultSet;
//
//
//public class MySQLSink1 extends RichSinkFunction<String> {
//    //private Connection connection=njuflink.KafkaFlinkMySQL.connection;
//    private PreparedStatement preparedStatement;
//    private String tb= GlobalConfig.TABLE;
//
//    @Override
//    public void open (org.apache.flink.configuration.Configuration parameters) throws Exception{
//
////        //加载JDBC驱动
////        Class.forName(njuflink.GlobalConfig.DRIVER_CLASS);
//        //获取数据库链接
////        connection= DriverManager.getConnection(njuflink.GlobalConfig.DB_URL,njuflink.GlobalConfig.USER_NAME,njuflink.GlobalConfig.PASSWORD);
////        // 关闭自定提交
////        connection.setAutoCommit(false);
//
//        System.out.println("sink db connected");
//
//    }
//    @Override
//    public void close() throws Exception{
//        if(preparedStatement!=null){
//            preparedStatement.close();
//        }
//        if(KafkaFlinkMySQL.connection!=null){
//            KafkaFlinkMySQL.connection.close();
//        }
//    }
//
//    public void invoke(String value,Context context)throws Exception{
//        if(value==null){
//            return;
//        }
//        System.out.println(value);
////        Collections.sort(new ArrayList<Map.Entry<String , Integer>>(njuflink.KafkaFlinkMySQL.lang2cnt.entrySet()),new Comparator<Map.Entry<String, Integer>>()
////        {
////            public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
//////                //按照value值升序
//////                return o1.getValue() - o2.getValue();
////                //按照value值降序
////                return o2.getValue() - o1.getValue();
////            }
////        });
//
//
//        String lang=value;
//
//        String sort=" order by count desc";
//        String querySQL="select * from "+tb+"where language = '"+lang+"';";
//        String updateSQL="update"+tb+"set count = count+1 "+" where language = '"+lang+"' ";
//        String insertSQL="insert into "+tb+"(language,count) values('"+lang+"', 1) ";
//
//        //查询数据
//        preparedStatement= KafkaFlinkMySQL.connection.prepareStatement(querySQL);
//        ResultSet resultSet=preparedStatement.executeQuery();
//        if(resultSet.next()){
//            //更新数据
//            //System.out.println(updateSQL);
//            preparedStatement.executeUpdate(updateSQL);
//        }else{
//            //插入数据
//            //System.out.println(insertSQL);
//            preparedStatement.execute(insertSQL);
//        }
//        // 提交
//        KafkaFlinkMySQL.connection.commit();
//
//
//    }
//
//}
//
////public class MySQLSink extends RichSinkFunction<Tuple2<String,Integer>> {
////    private Connection connection;
////    private PreparedStatement preparedStatement;
////    @Override
////    public void open (org.apache.flink.configuration.Configuration parameters) throws Exception{
////        super.open(parameters);
////        //加载JDBC驱动
////        Class.forName(njuflink.GlobalConfig.DRIVER_CLASS);
////        //获取数据库链接
////        connection= DriverManager.getConnection(njuflink.GlobalConfig.DB_URL,njuflink.GlobalConfig.USER_NAME,njuflink.GlobalConfig.PASSWORD);
////        System.out.println("sink db connect");
////
////    }
////    @Override
////    public void close() throws Exception{
////        super.close();
////        if(preparedStatement!=null){
////            preparedStatement.close();
////        }
////        if(connection!=null){
////            connection.close();
////        }
////        super.close();
////    }
////    @Override
////    public void invoke(Tuple2<String,Integer>value,Context context)throws Exception{
////        try{
////            String name=value.f0.replaceAll("[\\[\\]]","");
////            int count=value.f1;
////            String querySQL="select 1 from newscount "+"where name = "+name+"";
////            String updateSQL="update newscount set count = "+count+"where name = "+name+"";
////            String insertSQL="insert into newscount(name,count)values("+name+","+count+")";
////            //查询数据
////            preparedStatement=connection.prepareStatement(querySQL);
////            ResultSet resultSet=preparedStatement.executeQuery();
////            if(resultSet.next()){
////                //更新数据
////                preparedStatement.executeUpdate(updateSQL);
////            }else{
////                //插入数据
////                preparedStatement.execute(insertSQL);
////            }
////        }catch (Exception e){
////            e.printStackTrace();
////        }
////    }
////
////}
